/**
 * Copyright 2013 Nokia
 * All rights reserved
 */
package com.nokia.pc3.core.format.input;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import com.nokia.pc3.core.dac.NokiaUmem;
import com.nokia.pc3.core.xform.Transform;
import com.nokia.pc3.core.xform.Transformer;

/**
 * A record reader which makes NokiaRecords. The current key is just the line
 * offset, as with LineRecordReader.
 * 
 * @author michamur, jcroke
 * 
 */
public class NokiaUnstructuredRecordReader extends
		RecordReader<LongWritable, NokiaXFormRecord> {

	// Delegate most work to a simple RecordReader
	private RecordReader<LongWritable, Text> lr = null;
	private Transform transform = null;

	public NokiaUnstructuredRecordReader(
			RecordReader<LongWritable, Text> embeddedrr, Transform transform) {
		this.lr = embeddedrr;
		this.transform = transform;
	}

	@Override
	public void close() throws IOException {
		lr.close();
	}

	@Override
	public LongWritable getCurrentKey() throws IOException,
			InterruptedException {
		return lr.getCurrentKey();
	}

	@Override
	public NokiaXFormRecord getCurrentValue() throws IOException,
			InterruptedException {
		Text rawrecord = lr.getCurrentValue();

		if (null == rawrecord) {
			throw new IOException("input line is null");
		}

		transform.reset();
		NokiaUmem uMemRef = transform.getTransformUmem();
		// Add input for transformer. If > 1 transformer, first
		// one gets the input.
		if (transform.hasNext()) {
			transform.next();
			Transformer f = transform.getTransformerInstance();
			f.setInput(rawrecord.toString());
			f.run();
		}
		// If more transformers exist, run them.
		while (transform.hasNext()) {
			transform.next();
			Transformer f = transform.getTransformerInstance();
			f.run();
		}
		// Map uMem transformer outputs to oMem as defined by xformOutlist.
		uMemRef.setLabelOutput(transform.getXformation());
		// Swap the internal mem map with output.
		uMemRef.swap();
		NokiaXFormRecord nxf = new NokiaXFormRecord();
		nxf.init(uMemRef);
		return nxf;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		return lr.getProgress();
	}

	@Override
	public void initialize(InputSplit arg0, TaskAttemptContext arg1)
			throws IOException, InterruptedException {
		lr.initialize(arg0, arg1);

	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		return lr.nextKeyValue();
	}

}
